package kassiopeia.agent.internal.metrics;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import kassiopeia.agent.seda.config.SedaConnection;
import kassiopeia.plugin.Plugin;

public class MetricsAggregator {

    private static Boolean enabled = true;

    private static ConcurrentMap<String, Long> outgoingEventsCounter = new ConcurrentHashMap<String, Long>();
    private static ConcurrentMap<String, Long> processedEventsCounter = new ConcurrentHashMap<String, Long>();
    private static ConcurrentMap<String, Long> pluginProcessingErrorsCounter = new ConcurrentHashMap<String, Long>();

    public static void countOutgoingEvent(SedaConnection connection) {
        String stageId = connection.getFrom();
        incrementMapValue(outgoingEventsCounter, stageId);
    }

    public static void countProcessedEvent(Plugin plugin) {
        String pluginInstanceId = plugin.getContext().getPluginInstanceId();
        incrementMapValue(processedEventsCounter, pluginInstanceId);
    }

    public static void countPluginError(Plugin plugin) {
        String pluginInstanceId = plugin.getContext().getPluginInstanceId();
        incrementMapValue(pluginProcessingErrorsCounter, pluginInstanceId);
    }

    private static void incrementMapValue(ConcurrentMap<String, Long> map, String key) {
        Long oldValue = map.putIfAbsent(key, 1L);
        if (oldValue != null) {
            while (!map.replace(key, oldValue, oldValue + 1L)) {
                oldValue = map.get(key);
            }
        }
    }

    /**
     * returns number of all events send to specified stage
     */
    public static Long getOutgoingEventsCounterForStage(String stageId) {
        return outgoingEventsCounter.get(stageId);
    }

    /**
     * returns number of all events processed in specified stage
     */
    public static Long getProcessedEventsCounterForStage(String stageId) {
        return processedEventsCounter.get(stageId);
    }

    /**
     * returns number of all plugin errors in specified stage
     */
    public static Long getPluginErrorsCounterForStage(String stageId) {
        return pluginProcessingErrorsCounter.get(stageId);
    }

    public static void reset() {
        outgoingEventsCounter = new ConcurrentHashMap<String, Long>();
        processedEventsCounter = new ConcurrentHashMap<String, Long>();
        pluginProcessingErrorsCounter = new ConcurrentHashMap<String, Long>();
    }

    public static Boolean isEnabled() {
        return enabled;
    }
}
